package jmine.tec.subscription.impl;

import java.util.Collection;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

import jmine.tec.subscription.Publication;
import jmine.tec.subscription.Publisher;
import jmine.tec.subscription.SubscriptionManager;
import jmine.tec.subscription.SubscriptionSettings;
import jmine.tec.subscription.replicator.PublicationReplicator;
import jmine.tec.utils.io.IOUtils;

public class ThreadPoolReplicator implements PublicationReplicator {

    private final List<SubscriptionManagerImplementor> implementor = new CopyOnWriteArrayList<SubscriptionManagerImplementor>();

    private final ExecutorService executor = Executors.newFixedThreadPool(1, new ThreadFactory() {

        public Thread newThread(Runnable r) {
                Thread t = new Thread(r);
                t.setDaemon(true);
                return t;
            }});

    /**
     * {@inheritDoc}
     */
    public void setSubscriptionManager(SubscriptionManagerImplementor smi) {
        if (!this.implementor.contains(smi)) {
            this.implementor.add(smi);
        }
    }

    /**
     * {@inheritDoc}
     */
    public void register(Collection<? extends SubscriptionManagerImplementor> registers) {
        this.implementor.addAll(registers);
    }

    /**
     * {@inheritDoc}
     */
    public <P extends Publication> void synchronizePartialPublication(final Publisher<P> publisher, final SubscriptionSettings settings,
            PartialPublication<P> publication, SubscriptionManager source) {
        final PartialPublication<P> clone = IOUtils.deepClone(publication);
        // List<Callable<Void>> list = new ArrayList<Callable<Void>>();
        for (final SubscriptionManagerImplementor sm : this.implementor) {
            if (!source.equals(sm)) {
                final Callable<Void> task = new Callable<Void>() {
                    public Void call() throws Exception {
                        sm.localPublishPartialPublication(publisher, settings, clone);
                        return null;
                    }
                };
//                executor.submit(task);
                try {
                    task.call();
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }

    /**
     * {@inheritDoc}
     */
    public <P extends Publication> void synchronizePublication(final Publisher<P> publisher, final SubscriptionSettings settings,
            P publication,
            SubscriptionManager source) {
        final P clone = IOUtils.deepClone(publication);
        // List<Callable<Void>> list = new ArrayList<Callable<Void>>();
        for (final SubscriptionManagerImplementor sm : this.implementor) {
            if (!source.equals(sm)) {
                final Callable<Void> task = new Callable<Void>() {
                    public Void call() throws Exception {
                        sm.localPublish(publisher, settings, clone);
                        return null;
                    }
                };
//                executor.submit(task);
                try {
                    task.call();
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }

    public void awaitShutdown() {
        this.executor.shutdown();
        try {
            this.executor.awaitTermination(10000, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(e);
        }
    }

}
